package com.googlecode.pacswitch;

import java.io.IOException;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.Collection;
import java.util.concurrent.CopyOnWriteArrayList;

import com.googlecode.pacswitch.actions.Response;

/**
 * @author Andrey Pavlenko
 */
public class PacSwitch {
    private final Config _config;
    private final AsynchronousChannelGroup _channelGroup;
    private final Collection<CompletionHandler<Boolean, PacSwitch>> _listeners = new CopyOnWriteArrayList<>();
    private boolean _started;

    public PacSwitch(final Config config,
            final AsynchronousChannelGroup channelGroup) {
        _config = config;
        _channelGroup = channelGroup;
    }

    public Config getConfig() {
        return _config;
    }

    public AsynchronousChannelGroup getChannelGroup() {
        return _channelGroup;
    }

    public void addShutdownListener(
            final CompletionHandler<Boolean, PacSwitch> listener) {
        _listeners.add(listener);
    }

    public void removeShutdownListener(
            final CompletionHandler<Boolean, PacSwitch> listener) {
        _listeners.remove(listener);
    }

    public synchronized boolean isStarted() {
        return _started;
    }

    public synchronized void start() throws IOException {
        if (isStarted()) {
            throw new IllegalStateException();
        }

        final Config config = getConfig();
        final Log log = config.getLog();
        final AcceptHandler handler = new AcceptHandler();
        final AsynchronousChannelGroup channelGroup = getChannelGroup();
        log.init(this);
        log.info("Starting...");

        for (final Bind bind : config.getBindings()) {
            bind.bind(this, handler, channelGroup);
        }

        _started = true;
    }

    public void stop() {
        stop(null);
    }

    private synchronized void stop(final Throwable cause) {
        if (isStarted()) {
            final Log log = getConfig().getLog();
            log.info("Stopping...");
            _started = false;

            for (final Bind bind : getConfig().getBindings()) {
                bind.unbind(this);
            }

            if (cause != null) {
                for (final CompletionHandler<Boolean, PacSwitch> l : _listeners) {
                    l.failed(cause, this);
                }
            } else {
                for (final CompletionHandler<Boolean, PacSwitch> l : _listeners) {
                    l.completed(true, this);
                }
            }

            log.release(this);
        }
    }

    private final class AcceptHandler
            implements
            CompletionHandler<AsynchronousSocketChannel, AsynchronousServerSocketChannel> {

        @Override
        public void completed(final AsynchronousSocketChannel clientSocket,
                final AsynchronousServerSocketChannel ss) {
            final Config config = getConfig();
            final Log log = config.getLog();

            try {
                ss.accept(ss, this);
            } catch (final Throwable ex) {
                failed(ex, ss);
                return;
            }

            if (log.isDebugEnabled()) {
                log.debug("Incoming connection: " + clientSocket);
            }

            for (final Rule r : config.getRules()) {
                if (r.getCondition().apply(PacSwitch.this, clientSocket)) {
                    if (log.isInfoEnabled()) {
                        log.info("Applying rule: "
                                + r.getName().resolve(PacSwitch.this,
                                        clientSocket));
                    }

                    try {
                        r.getAction().perform(PacSwitch.this, clientSocket);
                    } catch (final Throwable ex) {
                        failed(ex, ss);
                    }

                    return;
                }
            }

            if (log.isWarnEnabled()) {
                log.warn("No matching rule found for: " + clientSocket);
            }

            try {
                final Response resp = new Response();
                resp.setStatusCode(new PInteger(500));
                resp.setReason(new PString("No matching rule found"));
                resp.perform(PacSwitch.this, clientSocket);
            } catch (final Throwable ex) {
                failed(ex, ss);
            }
        }

        @Override
        public void failed(final Throwable exc,
                final AsynchronousServerSocketChannel ss) {
            getConfig().getLog().error("Error occurred", exc);
            stop(exc);
        }
    }
}
